package main

import (
	"fmt"
	"os"
	"os/signal"
	"syscall"
	"time"

	"github.com/AsynkronIT/protoactor-go/actor"
	"github.com/louloulin/dataflare/pkg/actor/system"
	"github.com/louloulin/dataflare/pkg/actor/workflow"
)

func main() {
	fmt.Println("Starting DataFlare Actor Workflow Example")

	// Create actor system
	config := system.DefaultConfig()
	actorSystem, err := system.NewActorSystem(config)
	if err != nil {
		fmt.Printf("Failed to create actor system: %v\n", err)
		os.Exit(1)
	}

	// Get system supervisor
	systemSupervisor := actorSystem.SystemSupervisor

	// Get children
	future := actorSystem.System.Root.RequestFuture(systemSupervisor, &system.GetChildrenCommand{}, 5*time.Second)
	result, err := future.Result()
	if err != nil {
		fmt.Printf("Failed to get system supervisor children: %v\n", err)
		os.Exit(1)
	}

	response, ok := result.(*system.GetChildrenResponse)
	if !ok {
		fmt.Printf("Unexpected response type: %T\n", result)
		os.Exit(1)
	}

	// Get workflow supervisor
	workflowSupervisor := response.Children["workflow-supervisor"]
	if workflowSupervisor == nil {
		fmt.Println("Workflow supervisor not found")
		os.Exit(1)
	}

	// Create workflow definition
	workflowDefinition := &workflow.WorkflowDefinition{
		ID:          "example-workflow",
		Name:        "Example Workflow",
		Description: "An example workflow",
		Version:     "1.0.0",
		Sources: map[string]workflow.SourceDefinition{
			"memory-source": {
				Type: "memory",
				Config: map[string]interface{}{
					"data": []map[string]interface{}{
						{"id": "1", "value": 10},
						{"id": "2", "value": 20},
						{"id": "3", "value": 30},
					},
				},
			},
		},
		Processors: map[string]workflow.ProcessorDefinition{
			"transform-processor": {
				Type: "transform",
				Config: map[string]interface{}{
					"function": `
						function process(data) {
							const parsed = JSON.parse(data);
							parsed.value = parsed.value * 2;
							return JSON.stringify(parsed);
						}
					`,
				},
			},
		},
		Sinks: map[string]workflow.SinkDefinition{
			"console-sink": {
				Type: "console",
				Config: map[string]interface{}{
					"format": "json",
				},
			},
		},
		Edges: []workflow.EdgeDefinition{
			{
				From: workflow.NodeReference{SourceID: "memory-source"},
				To:   workflow.NodeReference{ProcessorID: "transform-processor"},
			},
			{
				From: workflow.NodeReference{ProcessorID: "transform-processor"},
				To:   workflow.NodeReference{SinkID: "console-sink"},
			},
		},
	}

	// Deploy workflow
	fmt.Println("Deploying workflow...")
	future = actorSystem.System.Root.RequestFuture(workflowSupervisor, &workflow.DeployWorkflowCommand{
		WorkflowID: "example-workflow",
		Definition: workflowDefinition,
	}, 5*time.Second)
	result, err = future.Result()
	if err != nil {
		fmt.Printf("Failed to deploy workflow: %v\n", err)
		os.Exit(1)
	}

	deployResponse, ok := result.(*workflow.DeployWorkflowResponse)
	if !ok {
		fmt.Printf("Unexpected response type: %T\n", result)
		os.Exit(1)
	}

	if !deployResponse.Success {
		fmt.Printf("Failed to deploy workflow: %s\n", deployResponse.Error)
		os.Exit(1)
	}

	fmt.Println("Workflow deployed successfully")

	// Start workflow
	fmt.Println("Starting workflow...")
	future = actorSystem.System.Root.RequestFuture(workflowSupervisor, &workflow.StartWorkflowCommand{
		WorkflowID: "example-workflow",
		Parameters: map[string]interface{}{
			"param1": "value1",
		},
	}, 5*time.Second)
	result, err = future.Result()
	if err != nil {
		fmt.Printf("Failed to start workflow: %v\n", err)
		os.Exit(1)
	}

	startResponse, ok := result.(*workflow.StartWorkflowResponse)
	if !ok {
		fmt.Printf("Unexpected response type: %T\n", result)
		os.Exit(1)
	}

	if !startResponse.Success {
		fmt.Printf("Failed to start workflow: %s\n", startResponse.Error)
		os.Exit(1)
	}

	runID := startResponse.RunID
	fmt.Printf("Workflow started with run ID: %s\n", runID)

	// Wait for workflow to complete or for user to interrupt
	fmt.Println("Workflow is running. Press Ctrl+C to stop...")
	sigCh := make(chan os.Signal, 1)
	signal.Notify(sigCh, syscall.SIGINT, syscall.SIGTERM)

	ticker := time.NewTicker(1 * time.Second)
	defer ticker.Stop()

	for {
		select {
		case <-ticker.C:
			// Get workflow status
			future = actorSystem.System.Root.RequestFuture(workflowSupervisor, &workflow.GetWorkflowStatusCommand{
				RunID: runID,
			}, 5*time.Second)
			result, err = future.Result()
			if err != nil {
				fmt.Printf("Failed to get workflow status: %v\n", err)
				continue
			}

			statusResponse, ok := result.(*workflow.GetWorkflowStatusResponse)
			if !ok {
				fmt.Printf("Unexpected response type: %T\n", result)
				continue
			}

			if !statusResponse.Success {
				fmt.Printf("Failed to get workflow status: %s\n", statusResponse.Error)
				continue
			}

			fmt.Printf("Workflow status: %s (Progress: %d%%)\n", statusResponse.Status, statusResponse.Progress)

			if statusResponse.Status == "completed" || statusResponse.Status == "failed" {
				fmt.Printf("Workflow %s\n", statusResponse.Status)
				return
			}

		case <-sigCh:
			fmt.Println("Stopping workflow...")
			future = actorSystem.System.Root.RequestFuture(workflowSupervisor, &workflow.StopWorkflowCommand{
				RunID: runID,
			}, 5*time.Second)
			result, err = future.Result()
			if err != nil {
				fmt.Printf("Failed to stop workflow: %v\n", err)
				os.Exit(1)
			}

			stopResponse, ok := result.(*workflow.StopWorkflowResponse)
			if !ok {
				fmt.Printf("Unexpected response type: %T\n", result)
				os.Exit(1)
			}

			if !stopResponse.Success {
				fmt.Printf("Failed to stop workflow: %s\n", stopResponse.Error)
				os.Exit(1)
			}

			fmt.Println("Workflow stopped successfully")
			
			// Shutdown actor system
			err = actorSystem.Shutdown()
			if err != nil {
				fmt.Printf("Failed to shutdown actor system: %v\n", err)
				os.Exit(1)
			}
			
			return
		}
	}
}
